package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;

import groupOne.bean.backUser;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Properties;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/1 9:13
 * @From: GoodGoodStudyDayDayUp
 *
 * 统计的是七日回流用户： 自上次登陆之后至少 7 日未登录的用户为回流用户
 *
 *    过滤数据：转成json对象格式方便取数据，然后应保留 uid 不为 null 且 last_page_id 不存在 或 last_page_id 为 login
 *
 *        根据用户分组 , 找出回流用户, 封装到POJO类里面：
 *    在状态中维护用户末次登陆日期，
 *    如果当天日期与末次登陆日期之差大于等于 8 天则回流用户数 backCt 置为 1
 *
 *          doris 建表语句：
 *                        drop table if exists dws_user_user_login_window;
 *                        create table if not exists dws_user_user_login_window
 *                        (
 *                            `stt`             DATETIME comment '窗口起始时间',
 *                            `edt`             DATETIME comment '窗口结束时间',
 *                            `cur_date`        DATE comment '当天日期',
 *                            `back_ct`         BIGINT replace comment '回流用户数',
 *                            `new_user`        BIGINT replace comment '新增用户数',
 *                            `active_user`     BIGINT replace comment '活跃用户数'
 *                        ) engine = olap aggregate key (`stt`, `edt`, `cur_date`)
 *                        comment "用户主题用户数汇总表"
 *                        partition by range(`cur_date`)()
 *                        distributed by hash(`stt`) buckets 10 properties (
 *                          "replication_num" = "3",
 *                          "dynamic_partition.enable" = "true",
 *                          "dynamic_partition.time_unit" = "DAY",
 *                          "dynamic_partition.start" = "-1",
 *                          "dynamic_partition.end" = "1",
 *                          "dynamic_partition.prefix" = "par",
 *                          "dynamic_partition.buckets" = "10",
 *                          "dynamic_partition.hot_partition_num" = "1"
 *                        );
 */
public class DwsUserComeBackUserWindow_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {
        new DwsUserComeBackUserWindow_lixuanfeng().init(3101,
                2,
                "DwsUserComeBackUserWindow",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.转换成jsonObject 且 过滤出所需的数据
        //TODO 此处代码测试OK
        SingleOutputStreamOperator<JSONObject> filterStream = parseAndFilter(stream);
            //filterStream.print();

         //2.找到7日回流用户的记录，累加起来封装到POJO类
        //TODO 此处代码测试OK
        SingleOutputStreamOperator<backUser> userAndCountPojoStream = findUserAndCount(filterStream);

        //3.开窗聚合
        //TODO 此处代码测试OK
        SingleOutputStreamOperator<backUser> AggStream = windowAgg(userAndCountPojoStream);
        //AggStream.print();

        //4.写入doris, Doris只支持 RowData数据 和 JSON 数据 写入, 写入前要先转换数据类型
        //TODO
        writeToDoris(AggStream);
    }

    private void writeToDoris(SingleOutputStreamOperator<backUser> aggStream) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_user_user_login_window";

        aggStream
                .map(pojo -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(pojo, config);
                })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));
        // 写入没测试
    }

    private SingleOutputStreamOperator<backUser> windowAgg(SingleOutputStreamOperator<backUser> userAndCountPojoStream) {
      return
        userAndCountPojoStream
                .assignTimestampsAndWatermarks( WatermarkStrategy
                        .<backUser>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner((b,ts)->b.getTs()))
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<backUser>() {
                            @Override
                            public backUser reduce(backUser value1, backUser value2) throws Exception {
                                value1.setNewUser(value1.getNewUser() + value2.getNewUser());
                                value1.setActiveUser(value1.getActiveUser() + value2.getActiveUser());
                                value1.setBackCt(value1.getBackCt() + value2.getBackCt());

                                return value1;
                            }
                        }
                        ,
                        new ProcessAllWindowFunction<backUser, backUser, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<backUser> elements, Collector<backUser> out) throws Exception {

                                backUser user = elements.iterator().next();

                                String start = AtguiguUtil.toDate(context.window().getStart());
                                String end = AtguiguUtil.toDate(context.window().getEnd());
                                String curdate = AtguiguUtil.toDate(System.currentTimeMillis());

                                user.setStt(start);
                                user.setEdt(end);
                                user.setCurDate(curdate);

                                out.collect(user);
                            }
                        }
                );

    }

    private SingleOutputStreamOperator<backUser> findUserAndCount(SingleOutputStreamOperator<JSONObject> filterStream) {
      return
        filterStream
                .keyBy(json->json.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, backUser>() {

                    private ValueState<Long> lastLoginDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastLoginDate = getRuntimeContext().getState(new ValueStateDescriptor<Long>("lastLoginDate",Long.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<backUser> out) throws Exception {
                        //首先看看状态是否为空，如果为空那就是第一次登录，新增用户记1
                        //如果不为空，看今天的日期和上次登录的日期之差是否大于7, 日期相减, 最好转换成时间戳，回流用户记1
                        //不是新增也不是回流, 即状态不为null 但是日期差值也不大于7，活跃用户记1

                        //新增用户，活跃用户，回流用户
                        Long newUsers =0L;
                        Long activeUsers=0L;
                        Long backUsers=0L;

                        Long lastDayTs = lastLoginDate.value();
                        Long todayTs = value.getLong("ts");

                        if (lastDayTs == null){
                            newUsers=1L;

                        }
                        else {
                            //如果差值小于7天
                            if ((todayTs - lastDayTs) < 7 * 1000 *60 * 60 * 24){

                                activeUsers=1L;
                                backUsers=0L;

                                //大于7天
                            }else{
                                activeUsers=0L;
                                backUsers=1L;
                            }
                        }

                        lastLoginDate.update(todayTs);
                    if (newUsers+activeUsers+backUsers>=1) {

                       // System.out.println(newUsers+""+activeUsers+""+backUsers);

                        out.collect(backUser.builder()
                                .newUser(newUsers)
                                .activeUser(activeUsers)
                                .backCt(backUsers)
                                .ts(todayTs)
                                .build());
                    } }
                });
    }


    private SingleOutputStreamOperator<JSONObject> parseAndFilter(DataStreamSource<String> stream) {
        return
            stream
                .map(JSON::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {

                        String uid = value.getJSONObject("common").getString("uid");
                        String lastPageId = value.getJSONObject("page").getString("last_page_id");

                        return  value.getJSONObject("common").containsKey("uid") &&
                            !value.getJSONObject("page").containsKey("last_page_id") &&
                                uid!=null && (lastPageId == null ||  "login".equals(lastPageId));
                    }
                });
    }


}
